package org.piggy.core.filter.router;

import com.alibaba.csp.sentinel.AsyncEntry;
import com.alibaba.csp.sentinel.SphU;
import com.alibaba.csp.sentinel.Tracer;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.alibaba.fastjson2.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.config.*;
import org.apache.dubbo.config.bootstrap.DubboBootstrap;
import org.apache.dubbo.rpc.service.GenericService;
import org.asynchttpclient.Request;
import org.asynchttpclient.Response;
import org.piggy.common.config.ServiceDefinition;
import org.piggy.common.constants.DubboConstants;
import org.piggy.common.constants.FilterConst;
import org.piggy.common.constants.GatewayProtocol;
import org.piggy.common.constants.OtherConst;
import org.piggy.common.enums.ResponseCode;
import org.piggy.common.exception.BreakerException;
import org.piggy.common.exception.ConnectException;
import org.piggy.common.exception.ResponseException;
import org.piggy.common.invoker.DubboServiceInvoker;
import org.piggy.common.invoker.ServiceInvoker;
import org.piggy.common.rule.Rule;
import org.piggy.common.utils.SystemUtil;
import org.piggy.core.ConfigLoader;
import org.piggy.core.context.GatewayContext;
import org.piggy.core.filter.Filter;
import org.piggy.core.filter.FilterAspect;
import org.piggy.core.dubbo.DubboParamResolveService;
import org.piggy.core.manager.DynamicConfigManager;
import org.piggy.core.netty.WebsocketClientHandler;
import org.piggy.core.reponse.GatewayResponse;
import org.piggy.core.request.GatewayRequest;
import org.piggy.core.utils.AsyncHttpUtil;
import org.piggy.core.utils.ResponseUtil;
import org.piggy.core.utils.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;

/**
 * 转发
 */
@FilterAspect(id= FilterConst.ROUTER_FILTER_ID, name = FilterConst.ROUTER_FILTER_NAME, order = FilterConst.ROUTER_FILTER_ORDER)
public class RouterFilter implements Filter {

    private static final Logger accessLog = LoggerFactory.getLogger("accessLog");

    private static final DubboBootstrap dubboBootstrap;

    static {
        ApplicationConfig applicationConfig = new ApplicationConfig();
        String applicationId = ConfigLoader.getConfig().getApplicationId();
        applicationConfig.setName(applicationId);
        applicationConfig.setQosEnable(false);

        String registryAddress = ConfigLoader.getConfig().getRegistryAddress();

        ConfigCenterConfig configCenterConfig = new ConfigCenterConfig();
        configCenterConfig.setAddress("nacos://" + registryAddress);

        RegistryConfig registryConfig = new RegistryConfig();
        registryConfig.setAddress("nacos://" + registryAddress);

        ProtocolConfig protocolConfig = new ProtocolConfig();
        protocolConfig.setName(CommonConstants.TRIPLE);
        protocolConfig.setPort(-1);

        MetadataReportConfig metadataReportConfig = new MetadataReportConfig();
        metadataReportConfig.setAddress("nacos://" + registryAddress);

        DubboBootstrap bootstrap = DubboBootstrap.getInstance();

        dubboBootstrap = bootstrap.metadataReport(metadataReportConfig).configCenter(configCenterConfig)
                .protocol(protocolConfig).application(applicationConfig).registry(registryConfig).start();

        accessLog.info("dubboBootstrap start finish");
    }

    @Override
    public void doFilter(GatewayContext gatewayContext) {
        try {
            AsyncEntry entry;
            Rule.CircuitBreakerConfig circuitBreakerConfig = gatewayContext.getRule().getCircuitBreakerConfig();
            if(circuitBreakerConfig != null){
                Integer count = circuitBreakerConfig.getCount();
                if(count != null && count > 0) {
                    entry = SphU.asyncEntry(OtherConst.DEGRADE + gatewayContext.getRule().getId());
                } else {
                    entry = null;
                }
            } else {
                entry = null;
            }

            String protocol = gatewayContext.getRule().getProtocol();
            if (GatewayProtocol.isDubbo(protocol)) {
                dubbo(gatewayContext, entry);
            } else {
                FullHttpRequest fullHttpRequest = gatewayContext.getGatewayRequest().getFullHttpRequest();
                String upgrade = fullHttpRequest.headers().get(HttpHeaderNames.UPGRADE);
                String connection = fullHttpRequest.headers().get(HttpHeaderNames.CONNECTION);
                // websocket
                // connection=Upgrade
                if ("Upgrade".equals(connection) && "websocket".equals(upgrade)) {
                    ThreadPoolUtil.getInstance().getExecutorService().execute(() -> websocket(gatewayContext, entry));
                } else {
                    // http
                    // connection=keep-alive
                    Request request = gatewayContext.getGatewayRequest().build();
                    CompletableFuture<Response> future = AsyncHttpUtil.getInstance().executeRequest(request);
                    boolean whenComplete = gatewayContext.getRule().getWhenComplete();
                    if (whenComplete) {
                        // 减少线程切换开销，性能更高
                        future.whenComplete((response, throwable) -> {
                            complete(response, throwable, gatewayContext, entry);
                        });
                    } else {
                        // 避免阻塞前序线程，适合耗时回调（如写数据库、发 HTTP 请求）
                        future.whenCompleteAsync((response, throwable) -> {
                            complete(response, throwable, gatewayContext, entry);
                        });
                    }
                }
            }
        } catch (BlockException e) {
            BreakerException breakerException = new BreakerException(ResponseCode.SERVICE_UNAVAILABLE);
            Tracer.trace(breakerException);
            accessLog.error("breaker -> ", breakerException);
            throw breakerException;
        }
    }

    private void dubbo(GatewayContext gatewayContext, AsyncEntry entry) {
        String serviceName = gatewayContext.getRule().getServiceName();
        ServiceDefinition serviceDefinition = DynamicConfigManager.getInstance().getServiceDefinition(serviceName);
        Map<String, ServiceInvoker> invokerMap = serviceDefinition.getInvokerMap();

        GatewayRequest request = gatewayContext.getGatewayRequest();
        String modifyPath = request.getModifyPath();

        DubboServiceInvoker serviceInvoker = (DubboServiceInvoker) invokerMap.get(modifyPath);

        ReferenceConfig<GenericService> reference = new ReferenceConfig<>();
        reference.setInterface(serviceInvoker.getInterfaceClass());
        reference.setVersion(serviceInvoker.getVersion());
        reference.setTimeout(serviceInvoker.getTimeout());
        reference.setGeneric("true");
        reference.setAsync(true);

        Rule.LoadBalanceConfig loadBalanceConfig = gatewayContext.getRule().getLoadBalanceConfig();
        String type;
        if (loadBalanceConfig == null) {
            type = FilterConst.LOAD_BALANCE_STRATEGY_RANDOM;
        } else {
            type = loadBalanceConfig.getType();
            if(type == null){
                type = FilterConst.LOAD_BALANCE_STRATEGY_RANDOM;
            }
        }
        switch (type) {
            case FilterConst.LOAD_BALANCE_STRATEGY_CONSISTENT_HASH ->
                    reference.setLoadbalance(DubboConstants.CONSISTENT_HASH);
            case FilterConst.LOAD_BALANCE_STRATEGY_WEIGHT, FilterConst.LOAD_BALANCE_STRATEGY_ROUND_ROBIN ->
                    reference.setLoadbalance(DubboConstants.ROUND_ROBIN);
            default ->
                    reference.setLoadbalance(DubboConstants.RANDOM);
        }

        String[] parameterTypes = serviceInvoker.getParameterTypes();

        ServiceLoader<DubboParamResolveService> serviceLoader = ServiceLoader.load(DubboParamResolveService.class);

        DubboParamResolveService dubboParamResolveService = serviceLoader.findFirst().orElse(new DubboParamResolveService() {
            @Override
            public Object[] buildParameter(String body, String[] parameterTypes) {
                Object[] objectArray = null;
                if (parameterTypes != null) {
                    JSONObject jsonObject = JSONObject.parseObject(body);
                    objectArray = new Object[]{jsonObject};
                }
                return objectArray;
            }
        });

        Object[] objectArray = dubboParamResolveService.buildParameter(request.getBody(), parameterTypes);

        GenericService genericService = dubboBootstrap.getCache().get(reference);
        CompletableFuture<Object> future = genericService.$invokeAsync(serviceInvoker.getMethodName(), parameterTypes, objectArray);
        boolean whenComplete = gatewayContext.getRule().getWhenComplete();
        if (whenComplete) {
            future.whenComplete(new BiConsumer<Object, Throwable>() {
                @Override
                public void accept(Object object, Throwable throwable) {
                    dubboComplete(object, throwable, gatewayContext, entry);
                }
            });
        } else {
            future.whenCompleteAsync(new BiConsumer<Object, Throwable>() {
                @Override
                public void accept(Object object, Throwable throwable) {
                    dubboComplete(object, throwable, gatewayContext, entry);
                }
            });
        }
    }

    private void websocket(GatewayContext gatewayContext, AsyncEntry entry) {
        try {
            GatewayRequest gatewayRequest = gatewayContext.getGatewayRequest();
            FullHttpRequest fullHttpRequest = gatewayContext.getGatewayRequest().getFullHttpRequest();
            String subProtocol = fullHttpRequest.headers().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL);
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(fullHttpRequest.uri(), subProtocol, true);

            ChannelHandlerContext ctx = gatewayContext.getChannelHandlerContext();
            WebSocketServerHandshaker obj = wsFactory.newHandshaker(fullHttpRequest);
            ChannelFuture future1 = obj.handshake(ctx.channel(), fullHttpRequest).sync();

            if(future1.isSuccess()){
                WebSocketClientHandshaker shaker = WebSocketClientHandshakerFactory
                        .newHandshaker(new URI(gatewayRequest.getWsUrl()),
                                WebSocketVersion.V13, subProtocol, true, fullHttpRequest.headers());

                WebsocketClientHandler websocketClientHandler = new WebsocketClientHandler(shaker);
                Bootstrap client = new Bootstrap();
                EventLoopGroup eventLoopGroup = AsyncHttpUtil.getInstance().getEventLoopGroup();
                client.group(eventLoopGroup)
                        .channel(useEpoll() ? EpollSocketChannel.class : NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) {
                                ChannelPipeline p = ch.pipeline();
                                p.addLast(new HttpClientCodec());
                                p.addLast(new HttpObjectAggregator(ConfigLoader.getConfig().getMaxContentLength()));
                                p.addLast(WebSocketClientCompressionHandler.INSTANCE);
                                p.addLast(websocketClientHandler);
                            }
                        });

                ChannelFuture future2 = client.connect(gatewayRequest.getIp(), gatewayRequest.getPort()).sync();
                if(future2.isSuccess()){
                    Channel channel1 = future1.channel();
                    Channel channel2 = future2.channel();
                    channel1.attr(AttributeKey.valueOf(OtherConst.FUTURE_2)).set(channel2);
                    channel2.attr(AttributeKey.valueOf(OtherConst.FUTURE_1)).set(channel1);
                }
                ReferenceCountUtil.release(fullHttpRequest);
            }
        }catch (Throwable t){
            accessLog.error("websocket -> ", t);
            gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
            gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(ResponseCode.INTERNAL_ERROR));
        }finally {
            if(entry != null){
                entry.close();
            }
        }
    }

    private boolean useEpoll() {
        return SystemUtil.isLinuxPlatform() && Epoll.isAvailable();
    }


    private void dubboComplete(Object object, Throwable throwable, GatewayContext gatewayContext, AsyncEntry entry) {
        try {
            if (throwable != null) {
                gatewayContext.setThrowable(new ResponseException(throwable, ResponseCode.DUBBO_RESPONSE_ERROR));
                gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(throwable.getMessage()));
            } else {
                gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(object));
            }
        } catch (Throwable t) {
            accessLog.error("dubbo throwable -> ", t);
            gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
            gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(ResponseCode.INTERNAL_ERROR));
        } finally {
            if(entry != null){
                entry.close();
            }
            gatewayContext.written();
            gatewayContext.releaseRequest();
            ResponseUtil.writeResponse(gatewayContext);
            accessLog.info("dubbo access message -> {}, {}, {}, {}, {}, {}, {}",
                    System.currentTimeMillis() - gatewayContext.getGatewayRequest().getBeginTime(),
                    gatewayContext.getRule().getServiceName(),
                    gatewayContext.getGatewayRequest().getClientIp(),
                    gatewayContext.getGatewayRequest().getMethod(),
                    gatewayContext.getGatewayRequest().getPath(),
                    gatewayContext.getGatewayResponse().getHttpResponseStatus().code(),
                    gatewayContext.getGatewayResponse().getResponse().getResponseBodyAsBytes().length);
        }
    }


    private void complete(Response response, Throwable throwable, GatewayContext gatewayContext, AsyncEntry entry) {
        try {
            if (throwable != null) {
                if (throwable instanceof TimeoutException) {
                    gatewayContext.setThrowable(new ConnectException(throwable, ResponseCode.REQUEST_TIMEOUT));
                    gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(ResponseCode.REQUEST_TIMEOUT));
                } else {
                    gatewayContext.setThrowable(new ResponseException(throwable, ResponseCode.HTTP_RESPONSE_ERROR));
                    gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(ResponseCode.HTTP_RESPONSE_ERROR));
                }
            } else {
                GatewayResponse gatewayResponse = ResponseUtil.buildGatewayResponse(response);
                gatewayContext.setGatewayResponse(gatewayResponse);
            }
        } catch (Throwable t) {
            accessLog.error("http throwable -> ", t);
            gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
            gatewayContext.setGatewayResponse(ResponseUtil.buildGatewayResponse(ResponseCode.INTERNAL_ERROR));
        } finally {
            if(entry != null){
                entry.close();
            }
            gatewayContext.written();
            gatewayContext.releaseRequest();
            ResponseUtil.writeResponse(gatewayContext);
            accessLog.info("http access message -> {}, {}, {}, {}, {}, {}, {}",
                    System.currentTimeMillis() - gatewayContext.getGatewayRequest().getBeginTime(),
                    gatewayContext.getRule().getServiceName(),
                    gatewayContext.getGatewayRequest().getClientIp(),
                    gatewayContext.getGatewayRequest().getMethod(),
                    gatewayContext.getGatewayRequest().getPath(),
                    gatewayContext.getGatewayResponse().getHttpResponseStatus().code(),
                    gatewayContext.getGatewayResponse().getResponse().getResponseBodyAsBytes().length);
        }
    }

}
